package com.rabbit.demo.direct;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

/**
 * @ClassName Producer
 * @Description TODO
 * @Author 11
 * @Date 2021/7/19 9:41
 * @Version 1.0
 */


@Service
@Slf4j
public class Producer {

    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * 消息返回监控
     * 启动消息失败就会返回，比如路由不到队列时触发
     *
     * @date 2021/7/19 10:49
     **/
    private final RabbitTemplate.ReturnsCallback returnsCallback = (returnedMessage) -> {
        String mqMessage = String.valueOf(returnedMessage.getMessage().getBody());
        String errorMsg = "Failed to send message,exchange:" + returnedMessage.getExchange()
                + ", routingKey:" + returnedMessage.getRoutingKey() + ", message :" + mqMessage
                + " , replyCode: " + returnedMessage.getReplyCode() + ", replyText:" + returnedMessage.getReplyText();
        log.error(errorMsg);
        /**
         * @Description 下面是错误后业务代码省略
         **/
    };

    private final RabbitTemplate.ConfirmCallback confirmCallback = (correlationData, ack, cause) -> {
        String correlationDataId = correlationData.getId();
        if (!ack) {/*如果没有确认*/
            log.error("消息没有被确认！");
        }else {
            log.info("消息被确认了:" + correlationDataId);
        }

    };



    /**
     *  将Java对象转换成amqp消息发送到mq上
     * <br/>
     * @param exchange	交换机
     * @param routingKey	路由key
     * @param msg	消息内容
     * @param correlationData	全局唯一消息id
     * @param priority	优先级
     * @param expiration 过期时间
     * @return void
     * @author ylr
     * @date 2021/7/19 13:01
     **/
    public void convertAndSend(String exchange, String routingKey, String msg,
                               CorrelationData correlationData, int priority, String expiration) {

        log.info("向mq推送的消息：【{}】 ",msg);

        rabbitTemplate.setConfirmCallback(confirmCallback);
        rabbitTemplate.setReturnsCallback(returnsCallback);
        MyMessgae myMessgae = new MyMessgae();
        myMessgae.setId("123");
        myMessgae.setObj("456");
        myMessgae.setAction("789");
        myMessgae.setStatus("111");
        myMessgae.setMqInfo(msg);



        rabbitTemplate.convertAndSend(exchange, routingKey, JSON.toJSONString(myMessgae),
                MessagePostProcessor -> {
                MessagePostProcessor.getMessageProperties().setPriority(priority);
                MessagePostProcessor.getMessageProperties().setExpiration(expiration);
                return MessagePostProcessor;
                },
                correlationData);


    }


}
